package com.shujia.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2CreateDF {

  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("df")
      .getOrCreate()


    /**
      * 1、 读取json文件
      *
      * json中自带列名，不需要指定列名
      *
      */

    val jsonDF: DataFrame = spark
      .read
      .format("json")
       .load("spark/data/students.json")


    /**
      * 2、读取文件格式的数据
      *
      */

    val csvDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",") //默认是逗号分隔
      //按照数据的顺序指定表结构
      .schema("id STRING , name STRING , age INT , gender STrING , clazz STRING")
      .load("spark/data/students.txt")

    csvDF.printSchema()
    //csvDF.show()


    /**
      *
      * 3、读取数据库构建df
      *
      */

    val jdbcDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "student.student")
      .option("user", "root")
      .option("password", "123456")
      .load()


    jdbcDF.show()

    /**
      * 保存一个parquet格式的文件
      *
      * parquet是一种带表结构的压缩格式
      *
      */

   // csvDF.write.mode(SaveMode.Overwrite).parquet("spark/data/parquet")


    /**
      * 4、读取parquet 格式的文件
      *
      */

  /*  val parquetDF: DataFrame = spark
      .read
      .format("parquet")
      .load("spark/data/parquet")

    parquetDF.printSchema()
    parquetDF.show()


    /**
      * 保存一个orc格式的数据
      *
      */

    csvDF.write.mode(SaveMode.Overwrite).orc("spark/data/orc")

    /**
      * 5、读取orc格式的数据
      * 是一种带表结构的压缩格式
      */


    val orcDF: DataFrame = spark
      .read
      .format("orc")
      .load("spark/data/orc")

    orcDF.printSchema()

    orcDF.show()


    /**
      * 6、基于RDD创建DF
      *
      */

    val sc: SparkContext = spark.sparkContext

    val rdd: RDD[String] = sc.textFile("spark/data/students.txt")

    val studentRDD: RDD[(String, String, Int, String, String)] = rdd.map(line => {
      val split: Array[String] = line.split(",")

      (split(0), split(1), split(2).toInt, split(3), split(4))
    })

    //导入隐式转换
    import spark.implicits._

    //将rdd转换成DF ,指定列名
    val studentDF: DataFrame = studentRDD.toDF("id", "name", "age", "gender", "clazz")
    studentDF.printSchema()
    studentDF.show()*/


    while (true){

    }

  }

}
